package concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import static java.lang.System.currentTimeMillis;

/**
 * 任务相关的：包含被执行的任务要实现的接口：Runnable接口或Callable接口
 * 任务的执行相关的：包含任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口。
 * Executor框架中有两个关键的类实现了ExecutorService接口（ThreadPoolExecutor和ScheduleThreadPoolExecutor）
 * 异步计算结果相关的：包含接口Future和实现Future接口的FutureTask类
 * ExecutorService有三种状态：运行、关闭、终止
 *
 * @author jack.wu
 * @since 2019/12/30 15:04
 */
public class ExecutorsDemo {

}


/**
 * ScheduleThreadPoolExecutor继承自ScheduleThreadPoolExecutor，他主要用来延迟执行任务，或者定时执行任务。
 * 功能和Timer类似，但是ScheduleThreadPoolExecutor更强大、更灵活一些。Timer后台是单个线程，
 * 而ScheduleThreadPoolExecutor可以在创建的时候指定多个线程。
 * schedule:延迟执行任务1次
 */
class ScheduleThreadPoolExecutorDemo {

    public static void main(String[] args) {
        ScheduledExecutorService service = Executors
                .newScheduledThreadPool(10, Executors.defaultThreadFactory());
        service.schedule(() -> {
            System.out.println("开始执行");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("执行结束");
        }, 5, TimeUnit.SECONDS);
    }
}

/**
 * scheduleAtFixedRate:固定的频率执行任务
 */
class ScheduleThreadPoolExecutorDemo1 {
    private static AtomicInteger count = new AtomicInteger(1);

    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService service = Executors
                .newScheduledThreadPool(10, Executors.defaultThreadFactory());
        ScheduledFuture<?> scheduledFuture = service.scheduleAtFixedRate(() -> {
            int countNum = count.getAndIncrement();
            System.out.println("第" + countNum + "执行");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("第" + countNum + "执行结束");
        }, 1, 1, TimeUnit.SECONDS);

        TimeUnit.SECONDS.sleep(5);
        // 判断任务是否已完成
        scheduledFuture.cancel(false);
        TimeUnit.SECONDS.sleep(1);
        System.out.println("任务是否被取消：" + scheduledFuture.isCancelled());
        System.out.println("任务是否已完成：" + scheduledFuture.isDone());
    }
}

/**
 * FutureTask除了实现Future接口，还实现了Runnable接口，因此FutureTask可以交给Executor执行，
 * 也可以交给线程执行执行（Thread有个Runnable的构造方法），FutureTask表示带返回值结果的任务。
 */
class FutureTaskDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask<Integer> futureTask = new FutureTask<>(() -> {
            TimeUnit.SECONDS.sleep(2);
            return 10;
        });
        futureTask.run();
        // new Thread(futureTask);
        System.out.println("结果：" + futureTask.get());
    }
}

/**
 * 例子：买新房了，然后在网上下单买冰箱、洗衣机，电器商家不同，所以送货耗时不一样，然后等他们送货，
 * 快递只愿送到楼下，然后我们自己将其搬到楼上的家中。
 */
class BuyGoodsDemo {

    static class GoodsModel {
        String name;
        long startTime;
        long endTime;

        public GoodsModel(String name, long startTime, long endTime) {
            this.name = name;
            this.startTime = startTime;
            this.endTime = endTime;
        }

        @Override
        public String toString() {
            return name + "，下单时间：[" + this.startTime + "],送达时间：[" + this.endTime + "]";
        }
    }

    /**
     * 搬运商品上楼
     *
     * @param goodsModel 商品信息
     * @throws InterruptedException InterruptedException
     */
    static void moveUp(GoodsModel goodsModel) throws InterruptedException {
        // 模拟将商品搬运上楼
        TimeUnit.SECONDS.sleep(3);
        System.out.println("将商品搬运上楼，商品信息：" + goodsModel);
    }

    /**
     * 模拟购买商品
     *
     * @param name     商品名称
     * @param costTime 耗时
     * @return 商品信息
     */
    static Callable<GoodsModel> buyGoods(String name, long costTime) {
        return () -> {
            long startTime = currentTimeMillis();
            System.out.println(startTime + "购买" + name);
            // 模拟送货耗时
            TimeUnit.SECONDS.sleep(costTime);
            long endTime = currentTimeMillis();
            System.out.println(endTime + "送达了" + name);
            return new GoodsModel(name, startTime, endTime);
        };
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        long startTime = currentTimeMillis();
        System.out.println("开始购物。。。");
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        Future<GoodsModel> cpFuture = executorService.submit(buyGoods("电脑", 5));
        Future<GoodsModel> wsFuture = executorService.submit(buyGoods("外设", 7));
        executorService.shutdown();
        // 等待电脑送到
        GoodsModel cpGoods = cpFuture.get();
        // 将电脑搬运上楼
        moveUp(cpGoods);
        // 等待外设送到
        GoodsModel wsGoods = wsFuture.get();
        // 将外设搬运上楼
        moveUp(wsGoods);
        long endTime = currentTimeMillis();
        System.out.println("完成整个购物，共耗时：" + (endTime - startTime));
    }
}

/**
 * 说一下其内部原理，ExecutorCompletionService创建的时候会传入一个线程池，调用submit方法传入需要执行的任务，
 * 任务由内部的线程池来处理；ExecutorCompletionService内部有个阻塞队列，任意一个任务完成之后，
 * 会将任务的执行结果（Future类型）放入阻塞队列中，
 * 然后其他线程可以调用它take、poll方法从这个阻塞队列中获取一个已经完成的任务，
 * 获取任务返回结果的顺序和任务执行完成的先后顺序一致，所以最先完成的任务会先返回。
 */
class ExecutorCompletionServiceDemo {

    static class GoodsModel {
        String name;
        long startTime;
        long endTime;

        public GoodsModel(String name, long startTime, long endTime) {
            this.name = name;
            this.startTime = startTime;
            this.endTime = endTime;
        }

        @Override
        public String toString() {
            return name + "，下单时间：[" + this.startTime + "],送达时间：[" + this.endTime + "]";
        }
    }

    /**
     * 搬运商品上楼
     *
     * @param goodsModel 商品信息
     * @throws InterruptedException InterruptedException
     */
    static void moveUp(GoodsModel goodsModel) throws InterruptedException {
        // 模拟将商品搬运上楼
        TimeUnit.SECONDS.sleep(3);
        System.out.println("将商品搬运上楼，商品信息：" + goodsModel);
    }

    /**
     * 模拟购买商品
     *
     * @param name     商品名称
     * @param costTime 耗时
     * @return 商品信息
     */
    static Callable<GoodsModel> buyGoods(String name, long costTime) {
        return () -> {
            long startTime = currentTimeMillis();
            System.out.println(startTime + "购买" + name);
            // 模拟送货耗时
            TimeUnit.SECONDS.sleep(costTime);
            long endTime = currentTimeMillis();
            System.out.println(endTime + "送达了" + name);
            return new GoodsModel(name, startTime, endTime);
        };
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        long startTime = currentTimeMillis();
        System.out.println("开始购物。。。");
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        ExecutorCompletionService<GoodsModel> completionService = new ExecutorCompletionService<>(executorService);
        completionService.submit(buyGoods("电脑", 5));
        completionService.submit(buyGoods("外设", 7));
        executorService.shutdown();
        int goodsCount = 2;
        for (int i = 0; i < goodsCount; i++) {
            // 获取最先到的商品
            GoodsModel goodsModel = completionService.take().get();
            moveUp(goodsModel);
        }
        long endTime = currentTimeMillis();
        System.out.println("完成整个购物，共耗时：" + (endTime - startTime));
    }
}

/**
 * 示例：执行一批任务，然后消费执行结果
 */
class ExecutorCompletionServiceDemo1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        int taskCount = 5;
        List<Callable<Integer>> tasks = new ArrayList<>(taskCount);
        for (int i = taskCount; i > 0; i--) {
            int j = i * 2;
            tasks.add(() -> j);
        }
        // 处理任务
        solve(executorService, tasks, c -> System.out.println(currentTimeMillis() + ":" + c));
    }

    private static <T> void solve(ExecutorService executorService, List<Callable<T>> tasks, Consumer<T> consumer) throws InterruptedException, ExecutionException {
        ExecutorCompletionService<T> completionService = new ExecutorCompletionService<>(executorService);
        for (Callable<T> t : tasks) {
            completionService.submit(t);
        }
        int size = tasks.size();
        for (int i = 0; i < size; i++) {
            T t = completionService.take().get();
            if (t != null) {
                consumer.accept(t);
            }
        }
    }
}

/**
 * 示例：异步执行一批任务，有一个完成立即返回，其他取消
 */
class ExecutorCompletionServiceDemo2 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        long startTime = currentTimeMillis();
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        int taskCount = 5;
        List<Callable<Integer>> tasks = new ArrayList<>(taskCount);
        for (int i = taskCount; i > 0; i--) {
            int j = i * 2;
            tasks.add(() -> j);
        }
        Integer value = invokeAny(executorService, tasks);
//        Integer value = executorService.invokeAny(tasks);
        long endTime = currentTimeMillis();
        executorService.shutdown();
        System.out.println("耗时：" + (endTime - startTime) + ",结果：" + value);
    }

    private static <T> T invokeAny(ExecutorService executorService, List<Callable<T>> tasks) throws InterruptedException, ExecutionException {
        ExecutorCompletionService<T> completionService = new ExecutorCompletionService<>(executorService);
        List<Future<T>> futures = new ArrayList<>(tasks.size());
        for (int i = 0; i < tasks.size(); i++) {
            Future<T> future = completionService.submit(tasks.get(i));
            futures.add(future);
        }
        try {
            for (int i = 0; i < tasks.size(); i++) {
                T t = completionService.take().get();
                if (t != null) {
                    return t;
                }
            }
        } finally {
            for (Future<T> future : futures) {
                future.cancel(true);
            }
        }
        return null;
    }


}
